package com.atguigu.flink.chapter08;

import com.atguigu.flink.bean.AdsClickLog;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

public class Flink06_High_Project_BlackList {
       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
               env.setParallelism(1);

            OutputTag<String> blacktag = new OutputTag<String>("black"){};
           SingleOutputStreamOperator<String> normalStream = env.readTextFile("input/AdClickLog.csv")
                   .map(line -> {
                       String[] datas = line.split(",");
                       return new AdsClickLog(
                               Long.valueOf(datas[0]),
                               Long.valueOf(datas[1]),
                               datas[2],
                               datas[3],
                               Long.valueOf(datas[4]));
                   })
                   //添加水印
                   .assignTimestampsAndWatermarks(
                           WatermarkStrategy
                                   .<AdsClickLog>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                   .withTimestampAssigner((log, ts) -> log.getTimestamp())
                   )
                   //每个用户对每个广告的点击量
                   .keyBy(log -> log.getUserId() + "_" + log.getAdsId())
                   .process(new KeyedProcessFunction<String, AdsClickLog, String>() {


                       private ValueState<Boolean> warnState;
                       private ReducingState<Long> countState;

                       @Override
                       public void open(Configuration parameters) throws Exception {
                           countState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<Long>(
                                   "countState", Long::sum, Long.class
                           ));
                           warnState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("warnState", Boolean.class));
                       }

                       @Override
                       public void processElement(AdsClickLog log,
                                                  Context ctx,
                                                  Collector<String> out) throws Exception {
                           //统计点击量的时候，只对那些没有加入黑名单的用户 进行统计
                          // warnState == null  不能这样写，因为其永远不为空,报空指针异常
                           if (warnState.value() ==null) {
                               countState.add(1L);
                           }

                           String msg = log.getUserId() + "对广告" + log.getAdsId() + " 的点击量是：" + countState.get();

                          //   out.collect(ctx.getCurrentKey() +" "+ countState.get());

                           if (countState.get() >99) {
                               if (warnState.value()==null) {
                                   ctx.output(blacktag,msg+" 放入黑名单");
                                   warnState.update(true);
                               }

                           }else {
                               out.collect(msg);}

                       }
                   });
           normalStream.print("正常用户");
           normalStream.getSideOutput(blacktag).print("黑名单");


           try {
                   env.execute();
               } catch (Exception e) {
                   e.printStackTrace();
               }


           }
}
